package com.talent.core.protocol.mqtt.handler;

import com.talent.core.protocol.mqtt.TlMqttSession;
import com.talent.core.protocol.mqtt.config.Constant;
import com.talent.core.protocol.mqtt.enums.TlMqttMessageType;
import com.talent.core.protocol.mqtt.enums.TlMqttQos;
import com.talent.core.protocol.mqtt.model.entity.TlSubClient;
import com.talent.core.protocol.mqtt.model.fix.TlMqttFixedHead;
import com.talent.core.protocol.mqtt.model.payload.TlMqttPublishPayload;
import com.talent.core.protocol.mqtt.model.request.TlMqttPubRelReq;
import com.talent.core.protocol.mqtt.model.request.TlMqttPublishReq;
import com.talent.core.protocol.mqtt.model.response.TlMqttPubComp;
import com.talent.core.protocol.mqtt.model.variable.TlMqttPubCompVariableHead;
import com.talent.core.protocol.mqtt.model.variable.TlMqttPubRelVariableHead;
import com.talent.core.protocol.mqtt.model.variable.TlMqttPublishVariableHead;
import com.talent.core.protocol.mqtt.store.IMqttStore;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;


/**
 * @ClassName：TlMqttRecHandler
 * @Author: hszhou
 * @Date: 2024/12/5 9:46
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
public class TlMqttPubRelHandler extends SimpleChannelInboundHandler<TlMqttPubRelReq> {

    private static final Logger logger = LoggerFactory.getLogger(TlMqttPubRelHandler.class);


    private IMqttStore mqttStore;

    public TlMqttPubRelHandler(IMqttStore mqttStore){
        this.mqttStore =mqttStore;
    }


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttPubRelReq req) throws Exception {
        logger.debug("进入【PUBREL】处理器");

        //根据这个消息获取到对应的
        TlMqttPubRelVariableHead variableHead = req.getVariableHead();
        Integer messageId = variableHead.getMessageId();

        //从缓存中拿到消息
        TlMqttPublishReq storeMsg = mqttStore.getStoreMsg(messageId);
        TlMqttPublishVariableHead vh = storeMsg.getVariableHead();
        TlMqttPublishPayload payload = storeMsg.getPayload();
        TlMqttFixedHead fixedHead = storeMsg.getFixedHead();

        String topic = vh.getTopic();
        int qos = fixedHead.getQos();
        int retain = fixedHead.getRetain();
        String content = payload.getContent().toString();
        int length = fixedHead.getLength();
        int messageQos = fixedHead.getQos();
        TlMqttQos mqttQos = TlMqttQos.valueOf(qos);
        List<TlSubClient> subClients = mqttStore.getSubClientByTopic(topic);

        //遍历订阅了这个主题的客户端 挨个发送
        for (TlSubClient subClient : subClients) {
            int subQos = subClient.getQos();
            TlMqttSession session = subClient.getSession();

            /*消息的订阅级别取最小的*/
            int readQos= Math.min(subQos, messageQos);
            TlMqttQos mqttQoS = TlMqttQos.valueOf(readQos);
            logger.info("转发qos2的消息向客户端客户端{}订阅主题{}的qos是{},发布消息的{}",subClient.getClientId(),subClient.getTopic(),subQos,messageQos);
            switch (mqttQoS){
                case QOS0:
                    TlMqttQos tlMqttQos = TlMqttQos.valueOf(messageQos);
                    if(tlMqttQos==TlMqttQos.QOS1||tlMqttQos==TlMqttQos.QOS2){
                        length-=Constant.SHORT_LENGTH;
                    }
                    forward(session,topic,mqttQoS,retain,content,length);
                    break;
                case QOS1:
                    forward(session, topic, mqttQoS, retain, content, fixedHead.getLength());
                    /*创建一个定时任务 判断是否收到了ack消息 如果没有就继续发送这 todo*/
                    break;
                case QOS2:
                    forward(session, topic, mqttQos, retain, content, length);
                    break;
                default:
                    break;

            }
        }
        sendComp(messageId,ctx);


    }

    /**
     * 发送comp消息给客户端 表示broker已经收到了消息
     * @param messageId
     * @param ctx
     */
    private void sendComp(int messageId,ChannelHandlerContext ctx){
        //说明服务端收到了这个消息
        mqttStore.removeStoreMsg(messageId);
        //发送comp消息
        TlMqttFixedHead head = new TlMqttFixedHead();
        head.setMessageType(TlMqttMessageType.PUBCOMP.getValue());
        head.setLength(Constant.SHORT_LENGTH);
        TlMqttPubCompVariableHead comVh = new TlMqttPubCompVariableHead();
        comVh.setMessageId(messageId);
        TlMqttPubComp res = new TlMqttPubComp(head,comVh);
        ctx.channel().writeAndFlush(res);
    }

    private TlMqttPublishReq forward(TlMqttSession session , String topic, TlMqttQos mqttQos, int retain, String content, int length){
        Integer identifierId = mqttStore.getMessageId();


        TlMqttPublishReq publishReq = new TlMqttPublishReq(topic, mqttQos, 0, retain,
                content,  identifierId, length);
        ChannelHandlerContext ctx = session.getCtx();
        //对于发送者 在发送前需要保存消息
        mqttStore.brokerStoreMsg(identifierId,publishReq);
        //如果转发的消息是qos2的并且是保留会话
        Set<TlMqttPublishReq> republish = session.getRepublish();
        republish.add(publishReq);
        if(null!=ctx){
            ctx.channel().writeAndFlush(publishReq);
            //todo 定时器n分钟没收到需要重新发送
            ctx.channel().eventLoop().schedule(()->{
                TlMqttPublishReq storeMsg = mqttStore.getStoreMsg(identifierId);
                if(storeMsg!=null){
                    ctx.channel().writeAndFlush(storeMsg);
                }
            },1, TimeUnit.MINUTES);
        }

        return publishReq;
    }
}
